package com.xcc.flink.hot

import java.sql.Timestamp

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer


//543462,1715,1464116,pv,1511658000 输入数据
case class UserBehavior(userId: Long, itemId: Long, categoryId: Long, behavior: String, timestamp: Long)

//统计格式输出类型
case class ItemViewCount(itemId: Long, windowEnd: Long, count: Long)

//统计每5分钟统计1小时内的topN
object FiveM1HourTopN {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //读取数据并解析为指定的UserBehavior对象
    val path = getClass.getResource("/UserBehavior.csv").getPath
    val textFileDStream: DataStream[UserBehavior] = env.readTextFile(path)
      .map {
        item => {
          val arr = item.split(",")
          UserBehavior(arr(0).trim.toLong, arr(1).trim.toLong, arr(2).trim.toLong, arr(3).trim, arr(4).trim.toLong)
        }
      }.filter(_.behavior == "pv") //按照pv进行统计

    //标记为升序,并指定eventTime的时间
    val sortEventDStream: DataStream[UserBehavior] = textFileDStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[UserBehavior](Time.seconds(0)) {
      override def extractTimestamp(element: UserBehavior): Long = element.timestamp * 1000
    })

    //先按照id聚合然后进行对应的时间滑动
    val resultDStream = sortEventDStream.keyBy(_.itemId).timeWindow(Time.hours(1), Time.minutes(5))
      //先获取到对应的ItemViewCount对象
      .aggregate(new CountAgg, new ResultAgg)
      //再根据窗口关闭的时间计算，最后得到对应这个时间窗口内的topN
      .keyBy(_.windowEnd)
      .process(new MyTopN(3))

    resultDStream.print()

    env.execute()
  }

}

//IN, ACC, OUT  输入为key 缓存为long  输出为long的数量
class CountAgg extends AggregateFunction[UserBehavior, Long, Long] {

  override def createAccumulator(): Long = 0L

  override def add(value: UserBehavior, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

//[IN, OUT, KEY, W <: Window]  IN输入为countAgg的输出long  out输出为ItemViewCount key为itemId的Long  W为timewindow
class ResultAgg extends WindowFunction[Long, ItemViewCount, Long, TimeWindow] {
  override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
    out.collect(ItemViewCount(key, window.getEnd, input.iterator.next()))
  }
}

//<K, I, O> key为Long类型的windowEnd  I为ItemViewCount 输出为字符串
case class MyTopN(topN: Int) extends KeyedProcessFunction[Long, ItemViewCount, String] {

  lazy val liststate: ListState[ItemViewCount] = getRuntimeContext.getListState[ItemViewCount](new ListStateDescriptor[ItemViewCount]("list-state", classOf[ItemViewCount]))

  override def processElement(value: ItemViewCount, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#Context, out: Collector[String]): Unit = {
    liststate.add(value)
    ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
    val listBuffer = new ListBuffer[ItemViewCount]
    val listIter = liststate.get().iterator()
    while (listIter.hasNext) {
      listBuffer += listIter.next()
    }
    //清空liststate
    liststate.clear()

    //对listBuffer进行排序
    val resultList = listBuffer.sortWith((x, y) => x.count > y.count).take(topN)

    val result: StringBuilder = new StringBuilder
    result.append("====================================\n")
    result.append("时间: ").append(new Timestamp(timestamp - 1)).append("\n")
    for(i <- resultList.indices){
      val currentItem: ItemViewCount = resultList(i)
      // e.g.    No1：    商品 ID=12224    浏览量=2413
      result.append("No").append(i+1).append(":")
        .append("    商品 ID=").append(currentItem.itemId)
        .append("    浏览量=").append(currentItem.count).append("\n")
    }
    result.append("====================================\n\n")
    //控制输出频率，模拟实时滚动结果
    Thread.sleep(1000)
    out.collect(result.toString)
  }

}
